package day07

import day07.caseClass.{CategoryCountInfo, UserVisitAction}
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2

import scala.collection.mutable
import scala.collection.mutable.ListBuffer

object TopNUtils {

  /**
   * 获取品类countRdd
   *
   * @param sc SparkContext
   * @return RDD[(String, CategoryCountInfo)] tuple._1：品类id；tuple._2：品类的一次操作行为(点击、下单、支付)
   */
  def getCategoryCountInfoRdd(sc: SparkContext): RDD[(String, CategoryCountInfo)] = {
    val actionRdd: RDD[String] = sc.textFile(path = "E:\\学习资料\\user_visit_action.txt")

    val userVisitActionRdd: RDD[UserVisitAction] = getUserVisitActionRdd(actionRdd)

    // 扁平化映射
    val categoryCountInfoRdd: RDD[(String, CategoryCountInfo)] = userVisitActionRdd.flatMap {
      userVisitAction => {
        // 如果是点击行为
        if (clickAction(userVisitAction)) {
          List(CategoryCountInfo(categoryId = userVisitAction.clickCategoryId.toString, clickCount = 1, orderCount = 0, payCount = 0))
        } else {
          // 如果是下单行为
          val categoryCountInfos: ListBuffer[CategoryCountInfo] = new ListBuffer[CategoryCountInfo]
          if (orderAction(userVisitAction)) {
            // 下单操作可能会有多个品类
            val categoryIds: Array[String] = userVisitAction.orderCategoryIds.split(",")
            for (categoryId <- categoryIds) {
              categoryCountInfos.append(CategoryCountInfo(categoryId = categoryId, clickCount = 0, orderCount = 1, payCount = 0))
            }
            categoryCountInfos
          } else {
            // 支付行为
            if (payAction(userVisitAction)) {
              // 支付操作可能会有多个品类
              val categoryIds: Array[String] = userVisitAction.payCategoryIds.split(",")
              for (categoryId <- categoryIds) {
                categoryCountInfos.append(CategoryCountInfo(categoryId = categoryId, clickCount = 0, orderCount = 0, payCount = 1))
              }
              categoryCountInfos
            } else {
              // 无效数据，返回空列表
              Nil
            }
          }
        }
      }
    }.map {
      categoryCountInfo => {
        val categoryId: String = categoryCountInfo.categoryId
        (categoryId, categoryCountInfo)
      }
    }

    categoryCountInfoRdd
  }

  /**
   *
   * 对RDD进行结构转换，拆解
   *
   * @param actionRdd RDD[String]
   * @return RDD[UserVisitAction]
   */
  def getUserVisitActionRdd(actionRdd: RDD[String]): RDD[UserVisitAction] = {
    // 数据结构转换，返回RDD[UserVisitAction]
    val userVisitActionRdd: RDD[UserVisitAction] = actionRdd.map {
      userVisitAction => {
        // 2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_37_2019-07-17 00:00:02_手机_-1_-1_null_null_null_null_3
        // 数据格式说明：
        //（1）数据采用_分割字段
        //（2）每一行表示用户的一个行为，所以每一行只能是四种行为中的一种。
        //（3）如果 searchKeyword 是null(这里的null是字符串)，表示这次不是搜索
        //（4）如果clickCategoryId和clickProductId是-1表示这次不是点击
        //（5）下单行为来说一次可以下单多个产品，所以品类id和产品id都是多个，id之间使用逗号，分割。如果本次不是下单行为，则他们相关数据用null来表示
        //（6）支付行为和下单行为类似
        val properties: Array[String] = userVisitAction.split("_")
        val date: String = properties(0)
        val userId: Long = properties(1).toLong
        val sessionId: String = properties(2)
        val pageId: Long = properties(3).toLong
        val actionTime: String = properties(4)
        val searchKeyword: String = properties(5)
        val clickCategoryId: Long = properties(6).toLong
        val clickProductId: Long = properties(7).toLong
        val orderCategoryIds: String = properties(8)
        val orderProductIds: String = properties(9)
        val payCategoryIds: String = properties(10)
        val payProductIds: String = properties(11)
        val cityId: Long = properties(12).toLong
        val action: UserVisitAction = UserVisitAction(date, userId, sessionId, pageId, actionTime, searchKeyword, clickCategoryId, clickProductId, orderCategoryIds, orderProductIds, payCategoryIds, payProductIds, cityId)
        action
      }
    }
    userVisitActionRdd
  }

  /**
   * 是否是点击行为
   *
   * @param userVisitAction <br>
   * @return true：点击行为；false：不是点击行为
   */
  def clickAction(userVisitAction: UserVisitAction): Boolean = {
    val clickCategoryId: Long = userVisitAction.clickCategoryId
    val clickProductId: Long = userVisitAction.clickProductId
    clickCategoryId != -1 && clickProductId != -1
  }

  /**
   * 是否是点击行为
   *
   * @param clickCategoryId 点击品类id
   * @param clickProductId  点击产品id
   * @return true：点击行为；false：不是点击行为
   */
  def clickAction(clickCategoryId: Long, clickProductId: Long): Boolean = {
    clickCategoryId != -1 && clickProductId != -1
  }

  /**
   * 是否是下单行为
   *
   * @param userVisitAction <br>
   * @return true：下单行为；false：不是下单行为
   */
  def orderAction(userVisitAction: UserVisitAction): Boolean = {
    val orderCategoryIds: String = userVisitAction.orderCategoryIds
    val orderProductIds: String = userVisitAction.orderProductIds
    !orderCategoryIds.equals("null") && !orderProductIds.equals("null")
  }

  /**
   * 是否是支付行为
   *
   * @param userVisitAction <br>
   * @return true：支付行为；false：不是支付行为
   */
  def payAction(userVisitAction: UserVisitAction): Boolean = {
    val payCategoryIds: String = userVisitAction.payCategoryIds
    val payProductIds: String = userVisitAction.payProductIds
    !payCategoryIds.equals("null") && !payProductIds.equals("null")
  }

  /**
   * 通过自定义累加器的方式获取Top10
   *
   * @param sc SparkContext
   * @return topN
   */
  def getTopNList(sc: SparkContext): List[CategoryCountInfo] = {

    val actionRdd: RDD[String] = sc.textFile(path = "E:\\学习资料\\user_visit_action.txt")

    // 对RDD进行结构转换，拆解
    val userVisitRdd: RDD[UserVisitAction] = TopNUtils.getUserVisitActionRdd(actionRdd)

    // 创建累加器
    val accumulator: CategoryCountAccumulator = new CategoryCountAccumulator

    // 注册累加器
    sc.register(accumulator, name = "CategoryCountAccumulator")

    // 使用累加器
    userVisitRdd.foreach(userVisitAction => accumulator.add(userVisitAction))

    // 获取累加器的值
    val accumulatorMap: mutable.Map[String, Long] = accumulator.value

    // 根据品类id进行分组

    // ((鞋-click),10)
    // ((鞋-order),20)
    // ((鞋-pay),30)
    val groupByMap: Map[String, mutable.Map[String, Long]] = accumulatorMap.groupBy {
      case (str, _) => str.split("-")(0)
    }

    val categoryCountInfos: List[CategoryCountInfo] = groupByMap.map {
      case (categoryId, categoryMap) =>

        val clickKey: String = categoryId.concat("-").concat("click")
        val clickCount: Long = categoryMap(clickKey)

        val orderKey: String = categoryId.concat("-").concat("order")
        val orderCount: Long = categoryMap(orderKey)

        val payKey: String = categoryId.concat("-").concat("pay")
        val payCount: Long = categoryMap(payKey)

        CategoryCountInfo(categoryId = categoryId, clickCount = clickCount, orderCount = orderCount, payCount = payCount)
    }.toList

    val topNList: List[CategoryCountInfo] = categoryCountInfos.sortWith((left, right) => sortRule(left, right)).take(n = 10)
    topNList
  }

  /**
   * 排序规则。按照点击数排名，靠前的就排名高；如果点击数相同，再比较下单数；下单数再相同，就比较支付数。
   *
   * @param left  <br>
   * @param right <br>
   * @return
   */
  private def sortRule(left: CategoryCountInfo, right: CategoryCountInfo): Boolean = {
    if (left.clickCount > right.clickCount) {
      return true
    }
    if (left.clickCount < right.clickCount) {
      return false
    }

    if (left.orderCount > right.orderCount) {
      return true
    }
    if (left.orderCount < right.orderCount) {
      return false
    }

    if (left.payCount > right.payCount) {
      return true
    }
    false
  }
}

/**
 * 自定义累加器。
 *
 * 输入：UserVisitAction；
 *
 * 输出：mutable.Map[String, Long]：key：categoryId+action拼接的字符串，中间用'-'连接; value：次数
 */
class CategoryCountAccumulator extends AccumulatorV2[UserVisitAction, mutable.Map[String, Long]] {

  private var map: mutable.Map[String, Long] = mutable.Map[String, Long]()

  /**
   * 返回此累加器是否为零值。 例如，对于计数器累加器，0为零值；对于列表累加器，Nil为零值。
   *
   * @return
   */
  override def isZero: Boolean = map.isEmpty

  /**
   * 创建此累加器的新副本。
   *
   * @return
   */
  def copy(): AccumulatorV2[UserVisitAction, mutable.Map[String, Long]] = {
    val accumulator: CategoryCountAccumulator = new CategoryCountAccumulator
    accumulator.map = map
    accumulator
  }

  /**
   * 复位该累加器，该累加器为零值。 即调用isZero必须返回true。
   */
  override def reset(): Unit = map.clear()

  /**
   * 接受输入并累积。
   *
   * @param userVisitAction 输入
   */
  def add(userVisitAction: UserVisitAction): Unit = {
    if (TopNUtils.clickAction(userVisitAction)) {
      val key: String = userVisitAction.clickCategoryId.toString.concat("-").concat("click")
      map(key) = map.getOrElse(key, 0L) + 1L
    } else if (TopNUtils.orderAction(userVisitAction)) {
      val orderCategoryIds: String = userVisitAction.orderCategoryIds
      val orderCategoryIdArr: Array[String] = orderCategoryIds.split(",")
      for (orderCategoryId <- orderCategoryIdArr) {
        val key: String = orderCategoryId.concat("-").concat("order")
        map(key) = map.getOrElse(key, 0L) + 1L
      }
    } else if (TopNUtils.payAction(userVisitAction)) {
      val payCategoryIds: String = userVisitAction.payCategoryIds
      val payCategoryIdArr: Array[String] = payCategoryIds.split(",")
      for (payCategoryId <- payCategoryIdArr) {
        val key: String = payCategoryId.concat("-").concat("pay")
        map(key) = map.getOrElse(key, 0L) + 1L
      }
    }
  }

  /**
   * 将另一个同类型的累加器合并到该累加器中并更新其状态，即该累加器应就地合并。
   *
   * @param other 另一个同类型的累加器
   */
  override def merge(other: AccumulatorV2[UserVisitAction, mutable.Map[String, Long]]): Unit = {
    val map1: mutable.Map[String, Long] = map
    val map2: mutable.Map[String, Long] = other.value
    map = map1.foldLeft(map2) {
      case (map, (key, value)) =>
        map(key) = map.getOrElse(key, 0L) + value
        map
    }
  }

  /**
   * 定义此累加器的当前值
   *
   * @return
   */
  def value: mutable.Map[String, Long] = map
}
